package com.yesun.task.pipeline.handler.impl;

import com.yesun.task.pipeline.common.Constant;
import com.yesun.task.pipeline.domain.dto.JobDto;
import com.yesun.task.pipeline.domain.dto.TaskDto;
import com.yesun.task.pipeline.handler.Executor;
import com.yesun.task.pipeline.handler.Handler;
import com.yesun.task.pipeline.handler.Loader;
import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;

/**
 * @Author yesun
 * @Descritpion
 * @Date 2023/8/17 6:36
 * @Version 1.0
 **/
@Slf4j
public class PythonHandler extends Handler {

    static SimpleDateFormat shortSdf = new SimpleDateFormat("yyyyMMdd");
    //从配置读取python路径，win和linux不同
    String pythonEnvPath = "E:\\Program Files\\python37\\python.exe";

    public PythonHandler(JobDto jobDto) {
        super(jobDto);
    }

    static Object lock = new Object();


    //通用PYTHON执行器

    @Override
    public Loader getLoader() {

        return new Loader() {
            @Override
            public void load(BlockingQueue queue) {
                //初始化python ETL任务，etl文件从外面传入
                log.info("加载PYTHON子任务...");
                TaskDto taskDto = new TaskDto();
                taskDto.setJobName(jobDto.getJobName());
                Map map = jobDto.getMap();
                List list = (List) map.get("params");
                if(list != null) {
                    taskDto.setName("PYTHON子任务execute file:" + list.get(0));
                }
                taskDto.setTag(Constant.JOB_TAG_PYTHON);
                //透传JOB中的map参数
                taskDto.setMap(jobDto.getMap());
                queue.offer(taskDto);
                log.info("添加子任务：" + taskDto.getName());
            }
        };
    }

    @Override
    public Executor getExecutor() {
        return new Executor() {
            @Override
            public void execute(TaskDto taskDto) {
                //执行python脚本
                log.info("执行Python脚本");
                //需要加锁
                synchronized (lock) {
                    log.info("取得锁，开始执行.....");
                    Map map = taskDto.getMap();
                    // 参数说明：第一个参数是执行器，这里是python的路径，若该路径已添加到环境变量，也可以直接写"python"，
                    // 第二个参数是python脚本路径，第三个及后面的参数是sys.argv的入参。所有参数都是String类型
                    //三个参数分别是，1、zjlr的目录，不支持子目录；2、保存目录；3、
                    ProcessBuilder processBuilder = new ProcessBuilder(pythonEnvPath);
                    if(map != null) {
                        List<String> params = (List<String>) map.get("params");
                        if(params != null) {
                            int i = 1;
                            for(String parameter : params) {
                                //map是无序的，这里会乱,所以改成list
                                processBuilder.command().add(i, parameter);
                                i++;
                            }
                        }
                    }
                    processBuilder.redirectErrorStream(true);
                    try {
                        Process process = processBuilder.start();
                        log.info(inputStreamToString(process.getInputStream()));
                        //执行完毕，如果有子任务，继续加入到Queue，暂时不支持 TODO，依赖问题最好用airflow
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
    }

    private static String inputStreamToString(InputStream inputStream) throws IOException {
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        byte[] tmp = new byte[1024];
        int len;
        while ((len = inputStream.read(tmp)) != -1) {
            outputStream.write(tmp, 0, len);
        }
        inputStream.close();
        outputStream.close();
        return outputStream.toString("GBK");
    }
}
